package udf;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import utils.MySqlTableUtils;
import utils.MySqlUtils;

import java.sql.Connection;
import java.sql.PreparedStatement;

/**
 * Create by zhh on 2022/9/9.
 * 往MySql去Sink的代码，只适用于Tuple2
 */
public class MySqlSinkT2 extends RichSinkFunction<Tuple2<String, Integer>> {
    Connection connection;
    PreparedStatement insertpsmt;
    PreparedStatement updatepsmt;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = MySqlUtils.getConnection();
        insertpsmt = connection.prepareCall("insert into "+ MySqlTableUtils.table_name +"("+MySqlTableUtils.col_1+","+MySqlTableUtils.col_2+") values (?,?)");
        updatepsmt = connection.prepareCall("update "+MySqlTableUtils.table_name+" set "+MySqlTableUtils.col_2+"=? where "+MySqlTableUtils.col_1+" =? ");
    }

    @Override
    public void close() throws Exception {
        if (connection!=null) connection.close();
        if (insertpsmt!=null) insertpsmt.close();
        if (updatepsmt!=null) updatepsmt.close();
    }

    @Override
    public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
        updatepsmt.setString(2, value.f0);
        updatepsmt.setInt(1,value.f1);
        updatepsmt.execute();
        if (updatepsmt.getUpdateCount()==0){
            insertpsmt.setString(1, value.f0);
            insertpsmt.setInt(2,value.f1);
            insertpsmt.execute();
        }
    }
}